package com.tmt.im.service.service;

import com.alibaba.fastjson2.JSONObject;
import com.tmt.im.common.enums.ChatSessionType;
import com.tmt.im.common.enums.ContentType;
import com.tmt.im.common.pojo.ChatDataKey;
import com.tmt.im.common.pojo.ChatDataPoJo;
import com.tmt.im.common.pojo.ChatSessionPoJo;
import com.tmt.im.common.pojo.IMUserPoJo;
import com.tmt.im.common.store.StoreSqlHelper;
import com.tmt.im.service.controller.vo.ChatDataQueryRequest;
import com.tmt.im.service.controller.vo.ChatDataVo;
import com.tmt.im.service.helper.PrincipalHelper;
import com.tmt.springboot.common.helper.IdSnowFlake;
import com.tmt.springboot.common.restful.ExecuteResult;
import com.tmt.springboot.common.restful.IExecuteResult;
import io.minio.GetPresignedObjectUrlArgs;
import io.minio.MinioClient;
import io.minio.PutObjectArgs;
import io.minio.RemoveObjectArgs;
import io.minio.http.Method;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.Ignite;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.security.Principal;
import java.sql.Timestamp;
import java.text.MessageFormat;
import java.time.OffsetDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

/**
 * @Description TODO
 * @Author wuyi
 * @Date 2024/8/11 9:32
 * @Version 1.0
 **/
@Slf4j
@Service
@Async("asyncTaskExecutor")
public class ChatDataService extends BaseService {

    private static final String[] ALLOW_FILE_TYPES = new String[]{};

    private final MinioClient minioClient;

    // Bucket 存储桶
    @Value("${minio.bucket}")
    private String bucket;

    private final IgniteBiPredicate<UUID, ChatDataKey> dataProcessor;

    private final Executor asyncTaskExecutor;

    public ChatDataService(MinioClient minioClient, Ignite ignite,
                            @Qualifier("dataProcessor")
                            IgniteBiPredicate<UUID, ChatDataKey> dataProcessor,
                            @Qualifier("asyncTaskExecutor")
                            Executor asyncTaskExecutor) {
        super(ignite);
        this.minioClient = minioClient;
        this.dataProcessor = dataProcessor;
        this.asyncTaskExecutor = asyncTaskExecutor;
    }

    /**
     * 将文件保存到minio中。
     *
     * @param dataVo
     * @param principal
     * @return
     */
    public CompletableFuture<IExecuteResult<Boolean>> appendChatData(ChatDataVo dataVo, Principal principal) {
        String url = null;
        String fileName = null;
        try {
            if (dataVo.getFile() != null) {
                fileName = dataVo.getFile().getOriginalFilename();
                assert fileName != null;

                //获取文件扩展名称。
                int index = fileName.lastIndexOf(".");
                String ext = index > 0 && index + 1 < fileName.length() ? fileName.substring(index) : "";
                if (ALLOW_FILE_TYPES.length > 0 && Arrays.stream(ALLOW_FILE_TYPES).noneMatch(x -> x.equals(ext))) {
                    throw new Exception("不支持上传的文件类型");
                }
                //获取随机文件名称
                String uid = UUID.randomUUID().toString().replaceAll("-", "");
                //文件名称。
                fileName = MessageFormat.format("ChatData/{1}{2}", uid, ext);
                //将文件保存到minio中。
                minioClient.putObject(PutObjectArgs.builder()
                        .bucket(this.bucket)        // 指定 Bucket
                        .contentType(MediaType.IMAGE_JPEG_VALUE)// 指定 Content Type
                        .object(fileName)        // 指定文件的路径
                        .stream(dataVo.getFile().getInputStream(), dataVo.getFile().getSize(), -1) // 文件的 Inputstream 流
                        .build());
                //获取文件访问路径。
                url = minioClient.getPresignedObjectUrl(GetPresignedObjectUrlArgs.builder()
                        .method(Method.GET)
                        .bucket(this.bucket)
                        .object(fileName)
                        .build());
            }
            String content = StringUtils.hasText(url) ? url : dataVo.getContent();
            Long id = IdSnowFlake.getId();
            ChatDataPoJo poJo = ChatDataPoJo.builder()
                    .id(id)
                    .mid(id)
                    .type(dataVo.getType())
                    .content(content)
                    .sessionId(dataVo.getSessionId())
                    .creatorId(PrincipalHelper.getCurrentUserId(principal))
                    .createTime(new Timestamp(System.currentTimeMillis()))
                    .insert(true)
                    .build();

            chatDataCache.put(poJo.getKey(), poJo);

            //在其他线程中完成用户消息推送
            CompletableFuture.runAsync(() -> {
                //向聊天会话中的用户推送消息
                sendOrdered(poJo);
            }, asyncTaskExecutor);

            return CompletableFuture.completedFuture(ExecuteResult.success("上传成功", true));
        } catch (Exception ex) {
            log.error(ex.getMessage(), ex);
            if (StringUtils.hasText(url)) {
                try {
                    //删除已存储的文件
                    minioClient.removeObject(RemoveObjectArgs.builder()
                            .bucket(this.bucket)
                            .object(fileName)
                            .build());
                } catch (Exception ex1) {
                }
            }
            return CompletableFuture.completedFuture(ExecuteResult.failure(ex.getMessage(), false));
        }
    }

    public CompletableFuture<IExecuteResult<Boolean>> revokeChatData(Long id, Principal principal) {
        try {
            String sql = MessageFormat.format("select * from {0} where mid = ?", ChatDataPoJo.CACHE_NAME);
            SqlFieldsQuery query = new SqlFieldsQuery(sql).setArgs(id);
            List<ChatDataPoJo> ChatDataPoJos = new ArrayList<>();
            try (FieldsQueryCursor<List<?>> cursor = chatDataCache.query(query)) {
                StoreSqlHelper.convertTo(cursor, ChatDataPoJos, ChatDataPoJo.class);
            }
            if (CollectionUtils.isEmpty(ChatDataPoJos)) {
                return CompletableFuture.completedFuture(ExecuteResult.success("要撤回的聊天记录不存在", false));
            }

            ChatDataPoJo dataPoJo = ChatDataPoJos.get(0);

            ChatSessionPoJo pojo = sessionCache.get(dataPoJo.getSessionId());
            if (pojo == null) {
                return CompletableFuture.completedFuture(ExecuteResult.failure("目标群组不存在", null));
            }

            Long curUserId = PrincipalHelper.getCurrentUserId(principal);
            //是否是群主
            boolean isOwner = isSessionOwner(curUserId, pojo);
            //是否是管理员
            boolean isManager = isManager(curUserId, dataPoJo.getSessionId());
            //只有群主和聊天记录的创建人才能撤回已推送的记录。
            if (!isOwner && !isManager && !Objects.equals(dataPoJo.getCreatorId(), curUserId)) {
                return CompletableFuture.completedFuture(ExecuteResult.success("无权撤回他人的聊天记录", false));
            }

            IMUserPoJo userPoJo = userCache.get(curUserId);

            //创建撤回记录。
            Long tmpId = IdSnowFlake.getId();
            ChatDataPoJo revoke = ChatDataPoJo.builder()
                    .id(tmpId)
                    .mid(tmpId)
                    .type(ContentType.REVOKE.name())
                    .content(MessageFormat.format("{0}撤回了一条消息",
                            Objects.equals(dataPoJo.getCreatorId(), curUserId) ? userPoJo.getNickname() : "群主"))
                    .sessionId(dataPoJo.getSessionId())
                    .referenceId(id)
                    .creatorId(curUserId)
                    .createTime(new Timestamp(System.currentTimeMillis()))
                    .insert(true)
                    .build();

            //修改被撤回的记录。
            dataPoJo.setDeleted(true);
            dataPoJo.setDeleteTime(revoke.getCreateTime());
            dataPoJo.setInsert(false);

            //创建乐观串行事务。
            try (Transaction tx = ignite.transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.SERIALIZABLE)) {
                try {
                    chatDataCache.replace(dataPoJo.getKey(), dataPoJo);
                    chatDataCache.put(new ChatDataKey(revoke.getId(), revoke.getSessionId()), revoke);
                    //提交事务
                    tx.commit();
                    return CompletableFuture.completedFuture(ExecuteResult.success("撤回成功", true));
                } catch (Exception ex) {
                    tx.rollback();
                    return CompletableFuture.completedFuture(ExecuteResult.failure(ex.getMessage(), false));
                }
            }
        } catch (Exception ex) {
            return CompletableFuture.completedFuture(ExecuteResult.failure(ex.getMessage(), false));
        }
    }

    /**
     * 查询聊天历史数据。
     *
     * @param queryRequest
     * @param principal
     * @return
     */
    public CompletableFuture<IExecuteResult<List<ChatDataPoJo>>> queryChatRecord(ChatDataQueryRequest queryRequest, Principal principal) {
        try {
            StringBuilder sb = new StringBuilder();
            sb.append("select * from " + ChatDataPoJo.CACHE_NAME + " ");
            sb.append("where sessionId = ? ");
            if (queryRequest.getStartId() != null) {
                if (queryRequest.isRollback()) {
                    sb.append("and mid < ? ");
                } else {
                    sb.append("and mid > ? ");
                }
            } else {
                sb.append("and mid != ? ");
                queryRequest.setStartId(0L);
            }

            if (StringUtils.hasText(queryRequest.getType())) {
                sb.append("and type = ? ");
            } else {
                sb.append("and type = ? ");
                queryRequest.setType(ContentType.TEXT.name());
            }

            if (StringUtils.hasText(queryRequest.getLike())) {
                sb.append("and content like (?) ");
                queryRequest.setLike(MessageFormat.format("%{0}%", queryRequest.getLike()));
            } else {
                sb.append("and content != ? ");
                queryRequest.setLike("");
            }

            if (queryRequest.isRollback()) {
                sb.append("order by mid desc ");
            } else {
                sb.append("order by mid asc ");
            }
            sb.append("limit ?");

            String sql = sb.toString();
            SqlFieldsQuery query = new SqlFieldsQuery(sql).setArgs(queryRequest.getSessionId(),
                    queryRequest.getStartId(), queryRequest.getType(), queryRequest.getLike(), queryRequest.getCount());
            List<ChatDataPoJo> ChatDataPoJos = new ArrayList<>();
            try (FieldsQueryCursor<List<?>> cursor = chatDataCache.query(query)) {
                StoreSqlHelper.convertTo(cursor, ChatDataPoJos, ChatDataPoJo.class);
            }
            return CompletableFuture.completedFuture(ExecuteResult.success("查询成功", ChatDataPoJos));
        } catch (Exception ex) {
            return CompletableFuture.completedFuture(ExecuteResult.failure(ex.getMessage(), null));
        }
    }

    private void sendOrdered(ChatDataPoJo pojo) {
        boolean send = dataProcessor.apply(null, pojo.getKey());
        ChatSessionPoJo sessionPoJo = sessionCache.get(pojo.getSessionId());
        if (!send || !sessionPoJo.getType().equalsIgnoreCase(ChatSessionType.DOUBLE.name())) {
            //通知远程节点分发数据。
            igniteMessaging.sendOrdered(pojo.getSessionId(), pojo.getKey(), 0);
        }
    }
}
